// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: leon

#include <experimental/filesystem>

#include <fstream>
#include <doctest.h>

#include "coin2/app/driver.h"
#include "coin2/app/driver.pb.h"
#include "coin2/base/config/app_config.h"
#include "coin2/exchange/base/feed/ws/subscriber_manager.h"
#include "coin2/exchange/di/constant.h"

using coin2::exchange::base::feed::SubscriberManager;
using coin2::exchange::di::ConstantInjectionBuilder;
using coin2::exchange::factory::feed::NewRequestGenerator;

std::vector<std::string> DeserializeAnswersFromPath(
  const std::experimental::filesystem::path& path) {
    std::vector<std::string> reqs;
    std::string line;
    auto file = std::ifstream(path);
    while (std::getline(file, line)) {
      reqs.push_back(line);
    }
    return reqs;
}

void SerializeAnswersToPath(
  std::vector<std::string> reqs,
  const std::experimental::filesystem::path& path) {
    std::ofstream ofs(path);
    for (const auto& req : reqs) {
      ofs << req << std::endl;
    }
    ofs.close();
}

std::vector<std::string> GenerateRestSubscribeReqs(
    const WsConnectionConfig& config) {
  std::map<std::string, std::set<std::string>> topic_requests;
  std::string http_string = "https://";
  int64_t interval = 0;
  std::string host;
  std::string port;
  std::string bind_ip;

  if (config.mea_config->mea.String() == "Futures.Huobi.v1") {
    return std::vector<std::string>{};
  }
  auto request_generator = NewRequestGenerator(config.mea_config->mea, config.ws_config);
  std::vector<std::string> reqs = request_generator->GenerateRequests(
      config.url.channels,
      config.symbols,
      config.mea_config->channels,
      config.url.request);
  CHECK_EQ(config.url.host.rfind(http_string), 0);

  int64_t num = config.url.unauthorized_limit.subscription->first;
  int64_t window = config.url.unauthorized_limit.subscription->second;
  interval = window / num + 1;
  std::string host_url = config.url.host.substr(http_string.size());
  host = host_url;
  port = config.url.port;
  bind_ip = config.url.bind_ip;
  std::stringstream ss;
  ss << "interval=" << interval << ",host=" << host << ",port=" << port << ",bind_ip=" <<bind_ip;
  reqs.emplace_back(ss.str());
  return reqs;
}

std::vector<std::string> GenerateWsSubscribeReqs(
  const WsConnectionConfig& config) {
    auto request_generator = NewRequestGenerator(config.mea_config->mea, config.ws_config);
    return request_generator->GenerateRequests(
        config.url.channels,
        config.symbols,
        config.mea_config->channels,
        config.mea_config->subscription_req);
}

std::vector<std::string> GenerateSubscribeReqs(
    const std::vector<WsConnectionConfig> configs) {
  std::vector<std::string> reqs;

  for (const WsConnectionConfig& config : configs) {
    if (config.url.host.rfind("https://") == 0) {
      // rest api starts with "https://"
      auto tmp = GenerateRestSubscribeReqs(config);
      reqs.insert(std::end(reqs), std::begin(tmp), std::end(tmp));
    } else {
      // default is websocket
      auto tmp = GenerateWsSubscribeReqs(config);
      reqs.insert(std::end(reqs), std::begin(tmp), std::end(tmp));
    }
  }
  return reqs;
}

void run(
    const std::experimental::filesystem::path& path,
    std::vector<std::string> answer) {
  coin2::app::DriverConfig config = coin2::base::config::DriverConfigFromJson(path);
  config.MergeFrom(coin2::base::config::DriverConfigFromJson(
      "data/coin2/feed/ops/feed-writer/worker1.json"));
  coin2::exchange::di::ConstantInjectionBuilder ci_builder;
  ci_builder.feed_symbol_groups_toml = "external/coin_test_data/subscribe/symbol_groups.toml";
  coin2::exchange::di::ConstantInjection ci(ci_builder);
  coin2::app::DriverConfig merged_config = coin2::app::ConsolidateDriverConfig(ci, &config);

  std::vector<WsConnectionConfig> ws_conn_configs;
  std::map<std::string, coin2::exchange::feed::FeedSystemConfig> config_map;
  for (const auto& [system_name, system_config] : merged_config.feed().exchanges()) {
    config_map.insert({system_name, system_config});
  }
  coin2::flow::SymbolGroupConfig::ResetInstance(
      "external/coin_test_data/subscribe/symbol_groups.toml");
  for (const auto& [system_name, system_config] : config_map) {
    auto mea = MarketExchangeApi::FromString(system_config.mea());
    SubscriberManager subscriber(
        ci.GetFeedSubscriptionSpec(mea),
        system_config.websocket(),
        coin2::flow::TopicType::LONG_TOPIC,
        nullptr,
        system_config.raw_parser().enable_telemetry());
    auto products = ProductRequestsFromConfig(
        mea.String(), system_config.products(), 1614947716 * 1e9);
    for (auto& prod : products) {
      auto symbol = prod->feed_native_symbol();
      char group = coin2::flow::SymbolGroupConfig::GetInstance().GetGroupForRelativeNorm(
          mea, prod->relative_norm());
      subscriber.RequestNativeSymbol(symbol, group);
    }
    auto tmp = subscriber.GenerateConnectionConfigs();
    ws_conn_configs.insert(std::end(ws_conn_configs), std::begin(tmp), std::end(tmp));
  }
  std::vector<std::string> generated_reqs = GenerateSubscribeReqs(ws_conn_configs);
  SerializeAnswersToPath(generated_reqs, "answer_dump");
  DOCTEST_CHECK(generated_reqs == answer);
  coin2::flow::SymbolGroupConfig::ResetInstance(
      "data/coin2/feed/symbol_groups.toml");
}

DOCTEST_TEST_CASE("feed subscribe") {
  DOCTEST_SUBCASE("default feed-05.ap-northeast-1.aws") {
    run("external/coin_test_data/subscribe/feed-05.ap-northeast-1.aws.json",
        DeserializeAnswersFromPath("external/coin_test_data/subscribe/answer"));
  }
}
